package org.harsh.transmission.client;

import cn.hutool.core.date.DateUtil;
import com.ruoyi.common.core.domain.entity.SysDept;
import com.ruoyi.system.service.ISysConfigService;
import io.jpower.kcp.netty.ChannelOptionHelper;
import io.jpower.kcp.netty.UkcpChannel;
import io.jpower.kcp.netty.UkcpChannelOption;
import io.jpower.kcp.netty.UkcpClientChannel;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedNioFile;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.handler.traffic.ChannelTrafficShapingHandler;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.harsh.common.Constants;
import org.harsh.dispatch.CommVars;
import org.harsh.dispatch.Dispatcher;
import org.harsh.dispatch.NodeStatusCheckTask;
import org.harsh.dispatch.TransmissionTask;
import org.harsh.transmission.common.GlobalTraffic;
import org.harsh.transmission.common.codec.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.*;
import java.util.Scanner;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
public class NodeClient {
    public static AtomicInteger REQID = new AtomicInteger(1);

    @Autowired
    private ISysConfigService sysConfigService;
    @Autowired
    private ClientChannelInitializer clientChannelInitializer;
    @Autowired
    private PlainMsgCodec plainMsgCodec;
    @Autowired
    private FileMsgCodec fileMsgCodec;

    // 发起节点连接
    public synchronized void connect(SysDept sysDept) {
        if (!CommVars.containNodeChannel(sysDept.getDeptId())
                && NodeStatusCheckTask.checkConnectivity(sysDept.getDeptId())) {
            if (sysDept.getLocalAddr() != null && !"".equals(sysDept.getLocalAddr())) {
                String addr = sysDept.getLocalAddr();
                int port = sysDept.getLocalPort();
                Bootstrap bootstrap = new Bootstrap();
                log.info("【正在建立连接】节点ID-{}, 节点地址-{}:{}", sysDept.getDeptId(), addr, port);
                bootstrap.group(CommVars.workGroup)
                        .channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY, true)
                        .handler(new LoggingHandler(LogLevel.INFO))
                        .handler(clientChannelInitializer);
                try {
                    ChannelFuture future = bootstrap.connect(addr, port).sync();
                    future.addListener(new ChannelFutureListener() {
                        // 连接建立完成后，将通道放入到Map中
                        public void operationComplete(ChannelFuture future) throws Exception {
                            if (future.isSuccess()) {
                                CommVars.nodeChannel.put(sysDept.getDeptId(), future.channel());
                                CommVars.addr2Node.put(sysDept.getLocalAddr(), sysDept.getDeptId());
                            }
                        }
                    });
                } catch (InterruptedException e) {
                    log.info("【建立连接失败】节点ID-{}, 节点地址-{}:{}", sysDept.getDeptId(), addr, port);
                    log.error(e.getLocalizedMessage());
                }

                Bootstrap kcpBootstrap = new Bootstrap();
                long channelWriteLimit = 16384L;
                long channelReadLimit = 16384L;
                String tmpWriteStr = sysConfigService.selectConfigByKey("trans.write.channelLimit");
                String tmpReadStr = sysConfigService.selectConfigByKey("trans.read.channelLimit");
                if(tmpWriteStr != null)
                    channelWriteLimit = Long.valueOf(tmpWriteStr);
                if(tmpReadStr != null)
                    channelReadLimit = Long.valueOf(tmpReadStr);
                long finalChannelWriteLimit = channelWriteLimit;
                long finalChannelReadLimit = channelReadLimit;
                kcpBootstrap.group(CommVars.workGroup).channel(UkcpClientChannel.class)
                        .handler(new ChannelInitializer<UkcpChannel>() {
                            @Override
                            public void initChannel(UkcpChannel ch) throws Exception {
                                ChannelPipeline pipeline = ch.pipeline();
                                pipeline.addLast(GlobalTraffic.getInstance())
                                        .addLast("channelTraffic",
                                                new ChannelTrafficShapingHandler(finalChannelWriteLimit, finalChannelReadLimit, 1))
                                        .addLast(new IdleStateHandler(30, 0, 0, TimeUnit.SECONDS))
                                        .addLast(new LengthFieldBasedFrameDecoder(14000, 0, 4, 0,4))
                                        .addLast(new LengthFieldPrepender(4))
                                        .addLast("cryptCodec", new CryptMsgPackCodec())
                                        .addLast("msgPackDecoder",new MsgPackDecoder())
                                        .addLast("msgPackEncoder",new MsgPackEncoder())
                                        .addLast("PlainMsgCodec", plainMsgCodec)
                                        .addLast("fileCodec", fileMsgCodec);
                            }
                        });
                ChannelOptionHelper.nodelay(kcpBootstrap, true, 20, 2, true)
                        .option(UkcpChannelOption.UKCP_MTU, 1400);
                try {
                    ChannelFuture future = kcpBootstrap.connect(addr, 8090).sync();
                    future.addListener(new ChannelFutureListener() {
                        // 连接建立完成后，将通道放入到Map中
                        public void operationComplete(ChannelFuture future) throws Exception {
                            if (future.isSuccess()) {
                                CommVars.nodeKcpChannel.put(sysDept.getDeptId(), future.channel());
                            }
                        }
                    });
                } catch (InterruptedException e) {
                    log.info("【建立Kcp连接失败】节点ID-{}, 节点地址-{}:{}", sysDept.getDeptId(), addr, 8090);
                    log.error(e.getLocalizedMessage());
                }
            }
        }
    }
}
